package spark_core
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.window
import org.apache.spark.{SparkConf, SparkContext}

import java.sql.Timestamp
import java.util.Properties

object Test2 {
    def main (args:Array[String]):Unit = {
      //    val conf = new SparkConf().setAppName("wordCountDemo")
      //    val sc = new SparkContext(conf)
      //    sc.textFile(args(0)).flatMap(line=>line.split(" "))
      //      .map(word=>(word,1))
      //      .reduceByKey((a,b)=>a+b,1)
      //      .saveAsTextFile(args(1))
      val spark = SparkSession.builder().appName("dateFrameDemo").getOrCreate()
      //    val csvDF = spark.read.format("csv")
      //      .option("header",true)
      //      .option("mode","FAILFAST")
      //      .load(args(2))
      //
      //    csvDF.show(3)
      //    csvDF.printSchema()
      //    csvDF.createGlobalTempView("fly")
      //    spark.sql("select * from fly where count>900").show(10)
      //    csvDF.filter()
      //    val url="jdbc:mysql://localhost:3306/spark?useUnicode=true&characterEncoding=utf-8&useSSL=false"
      //    val table="testdata"
      //    val properties=new Properties()
      //    properties.setProperty("user","root")
      //    properties.setProperty("password","123456")
      //    val sqlDF=spark.read.jdbc(url, table, properties)
      //    sqlDF.select("name","sex").distinct().show(5)
      import spark.implicits._
      var lines = spark.readStream.format("socket")
        .option("host","192.168.80.133")
        .option("port","9999")
        .option("includeTimeStamp",true)
        .load()

      val wordCount = lines.as[(String,Timestamp)]
        .flatMap(lines=>lines._1.split(" ").map(word=>(word,lines._2)))
        .toDF("word","timestamp")
        .groupBy(window($"timestamp","10 seconds","5 seconds"),$"word")
        .count()

      //    val words = wordCount.writeStream
      //      .format("console")
      //      .outputMode("complete")
      //      .start()
      val query = wordCount.writeStream
        .format("console")
        .outputMode("complete")
        .start()

      query.awaitTermination()
    }




}
